package com.atguigu.tingshu.album.listener;

import com.alibaba.fastjson.JSON;
import com.atguigu.tingshu.album.mapper.AlbumStatMapper;
import com.atguigu.tingshu.album.mapper.TrackStatMapper;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.model.album.AlbumStat;
import com.atguigu.tingshu.model.album.TrackStat;
import com.atguigu.tingshu.vo.album.TrackStatMqVo;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

/**
 * title:
 * author: zgl
 * date: 2024/3/8
 * description：
 */
@Component
public class AlbumListener {
    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private AlbumStatMapper albumStatMapper;

    @Autowired
    private TrackStatMapper trackStatMapper;

    @Transactional
    @KafkaListener(topics = KafkaConstant.TRACK_ALBUM_STAT_INCR)
    public void listen(String message) {
        TrackStatMqVo trackStatMqVo = JSON.parseObject(message, TrackStatMqVo.class);
        try {
            if (StringUtils.isBlank(message)) {
//                ack.acknowledge();
                return;
            }
            //幂等性校验 添加成功可以添加
            Boolean absent = this.redisTemplate.opsForValue().setIfAbsent(RedisConstant.TRACK_ALBUM_PLAY_NUM_PREFIX + trackStatMqVo.getBusinessNo(), "11111");
            if (!absent) {
//                ack.acknowledge();
                return;
            }
            //专辑表和声音表新增
            TrackStat trackStat = this.trackStatMapper.selectOne(new LambdaQueryWrapper<TrackStat>()
                    .eq(TrackStat::getTrackId, trackStatMqVo.getTarckId())
                    .eq(TrackStat::getStatType, trackStatMqVo.getStatType()));
            trackStat.setStatNum(trackStat.getStatNum() + trackStatMqVo.getCount());
            this.trackStatMapper.updateById(trackStat);


            String statType = null;
            if(SystemConstant.TRACK_STAT_PLAY.equals(trackStatMqVo.getStatType()) ||
            SystemConstant.TRACK_STAT_COMMENT.equals(trackStatMqVo.getStatType())){
                switch (trackStatMqVo.getStatType()){
                    case SystemConstant.TRACK_STAT_PLAY -> statType = SystemConstant.ALBUM_STAT_PLAY;
                    case SystemConstant.TRACK_STAT_COMMENT -> statType = SystemConstant.ALBUM_STAT_COMMENT;
                }
            }
            AlbumStat albumStat = this.albumStatMapper.selectOne(new LambdaQueryWrapper<AlbumStat>()
                    .eq(AlbumStat::getAlbumId, trackStatMqVo.getAlbumId())
                    .eq(AlbumStat::getStatType, statType));
            albumStat.setStatNum(albumStat.getStatNum() + trackStatMqVo.getCount());
            this.albumStatMapper.updateById(albumStat);
//            ack.acknowledge();
        } catch (Exception e) {
            this.redisTemplate.delete(RedisConstant.TRACK_ALBUM_PLAY_NUM_PREFIX + trackStatMqVo.getBusinessNo());
            throw new RuntimeException(e);
        }
    }
}
